package com.rocketmq.test.provider;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * @author hejianglong
 * @date 2019/12/6
 */
public class OrderedProducer {

    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("example_group_name");
        //Launch the instance.
        producer.setNamesrvAddr("10.211.55.10:9876");
        producer.start();
        String[] tags = new String[] {"TagA", "TagC", "TagD"};
        List<OrderDO> orderList = buildOrders();
        for (int i = 0; i < 10; i++) {
            int orderId = i % 10;
            //Create a message instance, specifying topic, tag and message body.
            String body = " Hello RocketMQ " + orderList.get(i).getOrderId() + orderList.get(i).getDesc();
            Message msg = new Message("TopicTest", tags[i % tags.length], "KEY" + i, body.getBytes(RemotingHelper.DEFAULT_CHARSET));

            SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long id = Long.parseLong(arg.toString());
                    long index = id % mqs.size();
                    return mqs.get((int)index);
                }
            }, orderId);

            System.out.printf("%s%n", sendResult);
        }
        //server shutdown
        producer.shutdown();
    }

    private static List<OrderDO> buildOrders() {
        List<OrderDO> orderList = new ArrayList<>();

        OrderDO OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111039");
        OrderDO.setDesc("创建");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111065");
        OrderDO.setDesc("创建");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111039");
        OrderDO.setDesc("付款");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103117235");
        OrderDO.setDesc("创建");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111065");
        OrderDO.setDesc("付款");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103117235");
        OrderDO.setDesc("付款");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111065");
        OrderDO.setDesc("完成");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111039");
        OrderDO.setDesc("推送");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103117235");
        OrderDO.setDesc("完成");
        orderList.add(OrderDO);

        OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111039");
        OrderDO.setDesc("完成");
        orderList.add(OrderDO);
        return orderList;
    }

    static class OrderDO implements Serializable {

        private String orderId;

        private String desc;

        public String getOrderId() {
            return orderId;
        }

        public void setOrderId(String orderId) {
            this.orderId = orderId;
        }

        public String getDesc() {
            return desc;
        }

        public void setDesc(String desc) {
            this.desc = desc;
        }
    }

}
